package com.shujia.flink.core

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

import java.time.Duration

object Demo6EventTime {
  def main(args: Array[String]): Unit = {

    /**
     * java,1683508875000
     * java,1683508876000
     * java,1683508877000
     * java,1683508879000
     * java,1683508880000
     * java,1683508878000
     * java,1683508881000
     * java,1683508882000
     * java,1683508885000
     * java,1683508890000
     */
    /**
     * 每隔5秒统计单词的数量，统计最近5秒的数据
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val wordsDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 解析数据，取出时间字段
     */
    val tsDS: DataStream[(String, Long)] = wordsDS.map(line => {
      val split: Array[String] = line.split(",")
      val word: String = split(0)
      val ts: Long = split(1).toLong
      (word, ts)
    })

    /**
     * 需要告诉flink哪一个字段是时间字段
     */
    // val assDS: DataStream[(String, Long)] = tsDS.assignAscendingTimestamps(kv => kv._2)

    /**
     * 指定时间字段和水位线
     * 水位线：最新数据的时间戳
     * 如果数据乱序了会导致数据的丢失，
     * 解决方法：将水位线前移，减去一段时间
     */
    val assDS: DataStream[(String, Long)] = tsDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //指定水位线前移的时间（数据最大乱序的时间）
        .forBoundedOutOfOrderness[(String, Long)](Duration.ofSeconds(5))
        //指定时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(element: (String, Long), recordTimestamp: Long): Long = element._2
        })
    )

    val kvDS: DataStream[(String, Int)] = assDS.map(kv => (kv._1, 1))


    /**
     * TumblingEventTimeWindows: 滚动的事件时间窗口
     */
    kvDS
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .sum(1)
      .print()


    env.execute()

  }

}
